package com.pass.zero;

/**
 * Created by IntelliJ IDEA.
 * User: yt347
 * Date: 5/7/11
 * Time: 5:05 PM
 * Converts all nodes to edge format
 */

import static com.utils.ConstantVariables.W_MAX;
import static com.utils.ConstantVariables.W_MIN;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import com.data.ConnectedEdge;
import com.data.Node;

/**
 * Pass Zero Mapper reads Nodes from the original data file, put the nodes in
 * appropriate groups and emits the format <GroupID, Edge> where Edge is the
 * Node points to itself.
 * 
 * @author bh349
 */
public class PassZeroMapper extends
		Mapper<LongWritable, Text, IntWritable, Text> {

	@Override
	public void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		StringTokenizer st = new StringTokenizer(value.toString());
		double[] vals = new double[3];
		for (int i = 0; st.hasMoreElements() && i < 3; i++) {
			vals[i] = Double.parseDouble(st.nextElement().toString());
		}
		if (W_MIN <= vals[2] && vals[2] <= W_MAX) {
			Node n = new Node(vals[0], vals[1]);
			ConnectedEdge edge = new ConnectedEdge(n, n);
			int[] gIds = n.getGroupNumbers();
			for (int gId : gIds) {
				if (gId != -1) {
					context.write(new IntWritable(gId),
							new Text(edge.toString()));
				}
			}
		}
	}
}
